/**
 * 2016-4-1 下午04:19:21
 */
package com.carpo.info;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.carpo.info.model.LazyStatus;
import com.carpo.info.model.PublishStatus;
import com.carpo.info.model.TInfos;
import com.carpo.info.service.IBaseInfoService;
import com.carpo.info.util.ServiceUtil;

public class Producer {
	
	
	private final static BlockingQueue<TInfos> queue = new LinkedBlockingDeque<TInfos>();
	private static final Logger logger = LoggerFactory.getLogger(Producer.class);
	
	public Producer() {
		try {
			//获取待发布和发布中的非延迟数据，避免项目重启导致数据无法处理
			//为避免项目重启，将DOING的延迟数据加载到队列中
			Map<String, IBaseInfoService> map = ServiceUtil.getServiceByType(IBaseInfoService.class);
			if(null != map){
				for(IBaseInfoService ser : map.values()){
					List<TInfos> list = ser.getTInfosByStatusAndLazy(PublishStatus.WAIT, LazyStatus.NO);
					if(null != list){
						queue.addAll(list);
					}
					List<TInfos> doList = ser.getTInfosByStatusAndLazy(PublishStatus.DOING, LazyStatus.NO);
					if(null != doList){
						queue.addAll(doList);
					}
					List<TInfos> lazyList = ser.getTInfosByStatusAndLazy(PublishStatus.DOING, LazyStatus.YES);
					if(null != lazyList){
						queue.addAll(lazyList);
					}
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		logger.info(".....start.....queue.length:" + queue.size());
		consume();
	}
	
	private void consume(){
		Thread thread = new Thread(new Runnable() {
			public void run() {
				while(true){
					try {
						TInfos info = queue.take();
						IBaseInfoService baseService = ServiceUtil.getServiceByTypeAndName(IBaseInfoService.class
								, info.getPlatForm());
						info.setPublishStatus(PublishStatus.DOING);
						baseService.updateTInfos(info);
						logger.info("......processing");
						String result = "";
						if(baseService.publishTInfos(info)){
							info.setPublishStatus(PublishStatus.SUCCESS);
							result = "success";
						}else{
							//非延迟消息如发布失败,则降为延迟消息再次发送
							info.setPublishStatus(PublishStatus.FAIL);
							info.setIsLazy(LazyStatus.YES);
							int count = info.getFailCount();
							count ++;
							info.setFailCount(count);
							result = "fail";
						}
						info.setPublishDate(new Date());
						baseService.updateTInfos(info);
						logger.info("......prosess complete : " + result + ", itemId : " + info.getItemId());
					} catch (Exception e) {
						e.printStackTrace();
						continue;
					}
				}
			}
		});
		thread.start();
	}
	
	public static void product(TInfos info){
		try {
			queue.put(info);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		logger.info("......put success......waiting process");
	}
	
}
